package mapReduce.homework.q3_orderAndMenuJoin;

import mapReduce.util.JobSubmit;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;

import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class OrderJoinMenuApp {

    public static class ojmMapper extends Mapper<LongWritable,Text,Text,NullWritable>{

        //初始化全局套餐map，封装小表
        Map<String,Double> menuMap = new HashMap<String, Double>();

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            //获得cacheFiles数组中的第一个文件menu.seq的URI路径
            URI menuURI = context.getCacheFiles()[0];
            //URI转成String类型的路径
            String menuFilePath = menuURI.getPath();

            //读取sequence类型文件数据
            Configuration conf = new Configuration();
            SequenceFile.Reader reader = new SequenceFile.Reader(FileSystem.get(conf), new Path(menuFilePath), conf);

            Text key = new Text();
            Text value = new Text();

            //遍历menu所有数据插入到map中
            while (reader.next(key, value)) {
                menuMap.put(key.toString(),Double.parseDouble(value.toString()));
            }
        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String line = value.toString();
            String[] fields = line.split("\t");

            //通过订单数据的最后一个字段内容找出相应套餐价格
            double money = menuMap.get(fields[fields.length - 1]);

            //原数据拼接新数据整体作为key，value为空
            context.write(new Text(line+"\t"+money),NullWritable.get());
        }
    }

    public static class ojmReducer extends Reducer<Text,NullWritable,Text,NullWritable>{
        @Override
        protected void reduce(Text key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            super.reduce(key, values, context);
        }
    }


    public static void main(String[] args) throws Exception {
        //生成sequence序列文件
//        Map<String,Object> map = new HashMap<String, Object>();
//        JobSubmitUtil.jobSubmitIO(args,OrderJoinMenuApp.class,SequenceFileOutputFormat.class);

        URI[] uris = {new URI("file:///E:/data_hadoop/homework/input/orderJoinMenu/menu.seq")};
        //生成文本文件
        JobSubmit.submitBaseJob(OrderJoinMenuApp.class,args,uris,SequenceFileOutputFormat.class);
    }
}
